package com.biconn.bigdata.jstrom.question.and.answer.pology;

import org.apache.commons.io.IOUtils;

import com.biconn.bigdata.jstrom.question.and.answer.bolt.QuestionAndAnswerBolt;
import com.biconn.bigdata.jstrom.question.and.answer.bolt.WordItemToKafkaBlot;
import com.biconn.bigdata.jstrom.question.and.answer.spout.QuestionAndAnswerSpout;
import com.biconn.bigdata.jstrom.util.LogbackInit;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;

public class QuestionAndAnswerLocalTopology {

	public static void main(String[] args) throws InterruptedException {
		if(args.length != 2) {
			throw new NullPointerException("请输入2个参数,第一个为运行模式,第二个为avro配置文件地址");
		}
		String toponame = "question-and-answer";
		LogbackInit.initLogback("logback.xml");
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("out-word", new QuestionAndAnswerSpout(),1);
		builder.setBolt("word-count", new QuestionAndAnswerBolt(),1).shuffleGrouping("out-word");
		builder.setBolt("word-count1", new WordItemToKafkaBlot(),1).shuffleGrouping("word-count");
		Config conf = new Config();
	    conf.setNumWorkers(1);
	    conf.setDebug(true);
	    conf.put("avro_path", args[1]);
	    conf.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);
        //本地模式:本地提交
        
        //cluster.killTopology("firstTopo");  
       // cluster.shutdown();  
		if(args[0].equals("0")) {
			LocalCluster cluster = new LocalCluster();  
	        cluster.submitTopology("firstTopo", conf, builder.createTopology());  
	        //一定要等待足够的时间，否则程序没来得及运行就已经结束了，程序启动需要消耗时间
	        Thread.sleep(Integer.MAX_VALUE);  
		}
		//远程模式
		try {
			StormSubmitter.submitTopologyWithProgressBar(toponame, conf, builder.createTopology());
		} catch (AlreadyAliveException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InvalidTopologyException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}


		
	}
	
	
	

}
